package cn.kingyea.bigdata.spark;
//创建SparkContext时所需引入的类。
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;

import cn.kingyea.utils.LoginUtil;
import scala.Tuple2;
import scala.Tuple3;

import org.apache.spark.api.java.JavaPairRDD;
//RDD操作时引入的类。
import org.apache.spark.api.java.JavaRDD;

import java.io.File;
import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
//创建SparkConf时引入的类。
import org.apache.spark.SparkConf;
/**
 * 服务器端没开。
 * @author hshe
 *
 */
public class SparkDemo {
	/**
	 * 所连接的集群认证模式是否为安全模式
	 */
	private static final boolean isSecureVerson = true;

	private static final String HIVE_DRIVER = "org.apache.hive.jdbc.HiveDriver";

	private static final String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client";
	private static final String ZOOKEEPER_SERVER_PRINCIPAL_KEY = "zookeeper.server.principal";
	private static final String ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL = "zookeeper/hadoop";

	private static Configuration CONF = null;
	private static String KRB5_FILE = null;
	private static String USER_NAME = null;
	private static String USER_KEYTAB_FILE = null;
	private static String zkQuorum = null;
	private static void init() throws IOException {
		CONF = new Configuration();
		// 其中，zkQuorum的"xxx.xxx.xxx.xxx"为集群中Zookeeper所在节点的IP，端口默认是24002
		zkQuorum = "192.168.0.13:24002,192.168.0.14:24002,192.168.0.15:24002";
		// 设置新建用户的USER_NAME，其中"xxx"指代之前创建的用户名，例如创建的用户为user，则USER_NAME为user
		USER_NAME = "dev_root";

		if (isSecureVerson) {
			// 设置客户端的keytab和krb5文件路径
			String userdir = System.getProperty("user.dir") + File.separator + "conf" + File.separator;
			USER_KEYTAB_FILE = userdir + "user.keytab";
			KRB5_FILE = userdir + "krb5.conf";
			
			CONF.addResource(new Path(userdir + "core-site.xml"));
			CONF.addResource(new Path(userdir + "hdfs-site.xml"));
			CONF.addResource(new Path(userdir + "hbase-site.xml"));
			
			LoginUtil.setJaasConf(ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, USER_NAME, USER_KEYTAB_FILE);
			LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_SERVER_PRINCIPAL_KEY, ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);

			// 安全模式
			// Zookeeper登录认证
			LoginUtil.login(USER_NAME, USER_KEYTAB_FILE, KRB5_FILE, CONF);
		}
	}
	
	
	
	
	
	
	
	
	
	public static void main(String[] args) throws Exception {
		init();
	    //创建一个配置类SparkConf，然后创建一个SparkContext
	    SparkConf conf = new SparkConf().setAppName("CollectFemaleInfo");
		
	    conf.setMaster("192.168.137.15");
		
	    JavaSparkContext jsc = new JavaSparkContext(conf);

	    //读取原文件数据,每一行记录转成RDD里面的一个元素
	    JavaRDD<String> data = jsc.textFile(args[0]);

	    //将每条记录的每列切割出来，生成一个Tuple
	    JavaRDD<Tuple3<String,String,Integer>> person = data.map(new Function<String,Tuple3<String,String,Integer>>()
	    {
	        private static final long serialVersionUID = -2381522520231963249L;

	        @Override
	        public Tuple3<String, String, Integer> call(String s) throws Exception
	        {
	            //按逗号分割一行数据
	            String[] tokens = s.split(",");

	            //将分割后的三个元素组成一个三元Tuple
	            Tuple3<String, String, Integer> person = new Tuple3<String, String, Integer>(tokens[0], tokens[1], Integer.parseInt(tokens[2]));
	            return person;
	        }
	    });

	    //使用filter函数筛选出女性网民上网时间数据信息    
	    JavaRDD<Tuple3<String,String,Integer>> female = person.filter(new Function<Tuple3<String,String,Integer>, Boolean>()
	    {
	        private static final long serialVersionUID = -4210609503909770492L;

	        @Override
	        public Boolean call(Tuple3<String, String, Integer> person) throws Exception
	        {
	            //根据第二列性别，筛选出是female的记录
	            Boolean isFemale = person._2().equals("female");
	            return isFemale;
	        }
	    });

	    //汇总每个女性上网总时间
	    JavaPairRDD<String, Integer> females = female.mapToPair(new PairFunction<Tuple3<String, String, Integer>, String, Integer>()
	    {
	        private static final long serialVersionUID = 8313245377656164868L;

	        @Override
	        public Tuple2<String, Integer> call(Tuple3<String, String, Integer> female) throws Exception
	        {
	            //取出姓名和停留时间两列，用于后面按名字求逗留时间的总和
	            Tuple2<String, Integer> femaleAndTime = new  Tuple2<String, Integer>(female._1(), female._3());
	            return femaleAndTime;
	        }
	    });
	      JavaPairRDD<String, Integer> femaleTime = females.reduceByKey(new Function2<Integer, Integer, Integer>()
	    {
	        private static final long serialVersionUID = -3271456048413349559L;

	        @Override
	        public Integer call(Integer integer, Integer integer2) throws Exception
	        {
	            //将同一个女性的两次停留时间相加，求和
	            return (integer + integer2);
	        }
	    });

	    //筛选出停留时间大于两个小时的女性网民信息
	    JavaPairRDD<String, Integer> rightFemales = females.filter(new Function<Tuple2<String, Integer>, Boolean>()
	    {
	        private static final long serialVersionUID = -3178168214712105171L;

	        @Override
	        public Boolean call(Tuple2<String, Integer> s) throws Exception
	        {
	            //取出女性用户的总停留时间，并判断是否大于2小时
	            if(s._2() > (2 * 60))
	            {
	                return true;
	            }
	            return false;
	        }
	    });

	    //对符合的female信息进行打印显示
	    for(Tuple2<String, Integer> d: rightFemales.collect())
	    {
	        System.out.println(d._1() + "," + d._2());
	    }
	}
}
